package com.group2.realtime.dwd.app;

import com.group2.edu.realtime.common.base.BaseSQLApp;
import com.group2.edu.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author 高耀
 * @date 2024/12/15 17:58<p></p>
 *
 * DWD层交易域支付成功事务事实表
 * 需要开启的程序：
 *       zk,kf,maxwell,flume1,DwdBaseLog,DwdTradeOrderDetail,DwdTradePaymentSuccessDetail
 */
public class DwdTradePaymentSuccessDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradePaymentSuccessDetail().start(
                10018,
                4,
                "DwdTradePaymentSuccessDetail"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment streamTableEnvironment) {
        //TODO 1 设置状态过期时间，下订单和支付之间有最大15分钟的业务间隔，再加10秒的数据传输延迟
        streamTableEnvironment.getConfig().setIdleStateRetention(Duration.ofSeconds(60 * 15 + 10));
        //TODO 2 先读取kafka_db中的数据，使用kafka的connector映射成表
        readTopicDb(streamTableEnvironment, "DwdTradePaymentSuccessDetail");

        //TODO 3 从topic_db中过滤出支付成功的信息
        //也就是table 为payment_info 、payment_status为 1602(成功)的数据
        Table paySuc = streamTableEnvironment.sqlQuery(
                "select " +
                        "data['order_id'] as order_id, " +
                        "data['alipay_trade_no'] as alipay_trade_no, " +
                        "data['payment_type'] as payment_type, " +
                        "data['payment_status'] as payment_status, " +
                        "data['callback_time'] as callback_time, " +
                        "data['callback_content'] as callback_content," +
                        "ts " +
                        "from topic_db " +
                        "where `table` = 'payment_info' " +
                        "and data['payment_status'] = '1602' "
        );
        //注册表名
        streamTableEnvironment.createTemporaryView("pay_suc", paySuc);

        //TODO 4 读取dwd的订单细节事务事实表数据
        streamTableEnvironment.executeSql(
                "create table dwd_trade_order_detail (" +
                        "id string,\n" +
                        "order_id string,\n" +
                        "course_id string,\n" +
                        "course_name string,\n" +
                        "user_id string,\n" +
                        "origin_amount string,\n" +
                        "coupon_reduce string,\n" +
                        "final_amount string,\n" +
                        "create_time string,\n" +
                        "province_id string,\n" +
                        "out_trade_no string,\n" +
                        "trade_body string,\n" +
                        "source_id string,\n" +
                        "ts bigint" +
                        ")" + SQLUtil.getKafkaProperty("dwd_trade_order_detail", "DwdTradePaymentSuccessDetail")
        );
        //TODO 5 关联两张表，以order_id为关联字段
        Table result = streamTableEnvironment.sqlQuery(
                "select " +
                        "`id`, " +
                        "od.order_id, " +
                        "course_id, " +
                        "course_name, " +
                        "user_id, " +
                        "origin_amount, " +
                        "coupon_reduce, " +
                        "final_amount, " +
                        "create_time as create_date, " +
                        "province_id, " +
                        "out_trade_no, " +
                        "trade_body, " +
                        "source_id, " +
                        "alipay_trade_no, " +
                        "payment_type, " +
                        "payment_status, " +
                        "callback_time, " +
                        "callback_content, " +
                        "od.ts " +
                        "from " +
                        "dwd_trade_order_detail od join pay_suc ps " +
                        "on od.order_id = ps.order_id "
        );
        //注册名字
        streamTableEnvironment.createTemporaryView("result_table", result);
        //TODO 6 将结果写入kafka对应topic
        //定义upsert-kafka连接器
        streamTableEnvironment.executeSql(
                "create table dwd_trade_payment_success_detail (" +
                        "id string,\n" +
                        "order_id string,\n" +
                        "course_id string,\n" +
                        "course_name string,\n" +
                        "user_id string,\n" +
                        "origin_amount string,\n" +
                        "coupon_reduce string,\n" +
                        "final_amount string,\n" +
                        "create_date string,\n" +
                        "province_id string,\n" +
                        "out_trade_no string,\n" +
                        "trade_body string,\n" +
                        "source_id string,\n" +
                        "alipay_trade_no string,\n" +
                        "payment_type string,\n" +
                        "payment_status string,\n" +
                        "callback_time string,\n" +
                        "callback_content string,\n" +
                        "ts bigint," +
                        "primary key(id) not enforced" +
                        ")" + SQLUtil.getUpsertKafkaProperty("dwd_trade_payment_success_detail")
        );
        //写入
        streamTableEnvironment.executeSql(
                "insert into dwd_trade_payment_success_detail select * from result_table");
    }
}
